package com.winning.xffw.service.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.winning.xffw.core.utils.CsvUtils;
import com.winning.xffw.core.utils.UUIDUtils;
import com.winning.xffw.service.entity.dto.BduDataLoadLog;
import com.winning.xffw.service.entity.dto.BduKafkaMsg;
import com.winning.xffw.service.entity.vo.TableDataVo;
import com.winning.xffw.service.service.IBduDataLoadLogService;
import com.winning.xffw.service.service.IBduKafkaMsgService;
import com.winning.xffw.service.service.IDataBaseService;
import com.winning.xffw.service.service.IDataLoadService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.IOException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DataLoadServiceImpl implements IDataLoadService {


    @Autowired
    private IDataBaseService dataBaseService;

    @Autowired
    private IBduKafkaMsgService bduKafkaMsgService;

    @Autowired
    private IBduDataLoadLogService bduDataLoadLogService;


    /**
     * 数据转储入口
     * 从table_data_msg中转储表数据
     */
    @Override
    public void dumpTableData() {
        //获取所有需要处理的表
        List<String> tableList = bduKafkaMsgService.queryTableList();
        log.info("==待处理的表[{}]", tableList);
        tableList.parallelStream().forEach(tableName -> {
            //按照表进行业务处理
            dumpTableDataByTable(tableName);
        });
    }

    /**
     * 按表进行数据转储操作
     *
     * @param tableName
     */
    private void dumpTableDataByTable(String tableName) {
        //取某表待处理的记录
        long startTime = System.currentTimeMillis();
        List<BduKafkaMsg> msgList = bduKafkaMsgService.queryDataMsgByTable(tableName);
        long endTime = System.currentTimeMillis();
        log.info("==获取表[{}]待处理记录数[{}]完成，耗时{}ms。", tableName, msgList.size(), endTime - startTime);
        msgList.forEach(msg -> {
            try {
                log.info("==数据表[{}]、Kafkid[{}]执行开始", msg.getBm(), msg.getKafkid());
                long startTimeMillis = System.currentTimeMillis();
                doExecute(msg);
                long endTimeMillis = System.currentTimeMillis();
                log.info("==数据表[{}]、Kafkid[{}]执行结束，耗时{}ms",
                        msg.getBm(), msg.getKafkid(), (endTimeMillis - startTimeMillis));
            } catch (Exception e) {
                log.error("==消息处理失败：kafkid=" + msg.getKafkid(), e);
            }
        });
    }


    /**
     * 从table_data_msg中转储表数据
     *
     * @param bduKafkaMsg
     */
    private void doExecute(BduKafkaMsg bduKafkaMsg) {
        //日志
        BduDataLoadLog bdlLog = new BduDataLoadLog();
        long startTimeMillis = System.currentTimeMillis();
        boolean result = true;
        try {
            //校验入参信息
            checkParam(bduKafkaMsg);
            TableDataVo tableDataVo = convertMsgData(bduKafkaMsg);
            //目标表数据批量更新
            batchUpdateTargetTable(tableDataVo);
            //目标表数据批量导入
            long successCount = copyData2TargetTable(tableDataVo);
            bdlLog.setZsl(tableDataVo.getDataRows().size());
            bdlLog.setCgsl((int) successCount);
        } catch (Exception e) {
            result = false;
            bdlLog.setRzxx(e.getMessage() + "\n" + ExceptionUtils.getStackTrace(e));
            log.error("数据数据转储失败！", e);
        }
        long endTimeMillis = System.currentTimeMillis();
        //5、处理状态更新
        bduKafkaMsgService.updateStatus(bduKafkaMsg.getKafkid(), result);
        //记录日志
        bdlLog.setUid(UUIDUtils.getUUID());
        bdlLog.setKafkid(bduKafkaMsg.getKafkid());
        bdlLog.setRwid(bduKafkaMsg.getRwid());
        bdlLog.setBm(bduKafkaMsg.getBm());
        bdlLog.setJgbm(bduKafkaMsg.getJgbm());
        bdlLog.setZt(result ? "1" : "0");
        bdlLog.setClkssj(new Timestamp(startTimeMillis));
        bdlLog.setCljssj(new Timestamp(endTimeMillis));
        bduDataLoadLogService.saveDataLoadLog(bdlLog);
    }

    /**
     * 目标表数据批量更新
     *
     * @param tableDataVo
     */
    private void batchUpdateTargetTable(TableDataVo tableDataVo) {
        String basePath = getBasePath(tableDataVo.getTargetTableName());
        String filename = basePath + tableDataVo.getKafkId() + "_temp.csv";
        //1、导出数据到csv
        try {
            CsvUtils.list2Csv(filename, tableDataVo.getIndexColumns(), tableDataVo.getDataRows());
        } catch (IOException e) {
            log.error("数据转储csv文件失败！", e);
            throw new RuntimeException(e);
        }
        //2、通过临时表批量更新目标表
        dataBaseService.batchUpdateTargetTableByTemp(filename, tableDataVo);
        new File(filename).delete();
    }

    /**
     * 校验信息
     *
     * @param bduKafkaMsg
     */
    private void checkParam(BduKafkaMsg bduKafkaMsg) throws Exception {
        //校验目标表是否存在
        boolean isExists = dataBaseService.checkTableIsExists(bduKafkaMsg.getBm());
        if (!isExists) {
            log.error("目标表{}不存在", bduKafkaMsg.getBm());
            throw new Exception("目标表" + bduKafkaMsg.getBm() + "不存在!");
        }

    }

    /**
     * 数据插入至目标表
     *
     * @param tableDataVo
     */
    private long copyData2TargetTable(TableDataVo tableDataVo) {
        String basePath = getBasePath(tableDataVo.getTargetTableName());
        String filename = basePath + tableDataVo.getKafkId() + "_target.csv";
        //从首行数据获得headers
        List<Map<String, Object>> dataRows = tableDataVo.getDataRows();
        if (dataRows == null || dataRows.isEmpty()) {
            log.info("dataRows数据为空！");
            return 0;
        }
        List<String> headers = dataRows.get(0).keySet().stream().collect(Collectors.toList());
        //1、导出数据到csv
        try {
            CsvUtils.list2Csv(filename, headers, tableDataVo.getDataRows());
        } catch (IOException e) {
            log.error("数据转储csv文件失败！", e);
            throw new RuntimeException(e);
        }
        //2、从csv导数据至数据表
        long successCount = dataBaseService.copyCsvData2Table(filename, tableDataVo.getTargetTableName(),
                headers, tableDataVo.getDataRows().size());
        new File(filename).delete();
        return successCount;
    }

    /**
     * 返回存储csv文件路径
     *
     * @return
     */
    private String getBasePath(String tableName) {
        String basePath = System.getProperty("user.dir") + File.separator + "tempData" + File.separator
                + new SimpleDateFormat("yyyyMMdd").format(new Date()) + File.separator
                + tableName + File.separator;
        new File(basePath).mkdirs();
        return basePath;
    }

    /**
     * 对象转换
     *
     * @param bduKafkaMsg
     * @return
     */
    private TableDataVo convertMsgData(BduKafkaMsg bduKafkaMsg) {
        TableDataVo vo = new TableDataVo();
        vo.setKafkId(bduKafkaMsg.getKafkid());
        vo.setRwId(bduKafkaMsg.getRwid());
        vo.setJgbm(bduKafkaMsg.getJgbm());
        vo.setTargetTableName(bduKafkaMsg.getBm());
        vo.setTempTableName(bduKafkaMsg.getBm() + "_temp");
        vo.setIndexColumns(Arrays.asList(bduKafkaMsg.getZdm().split("\\|")));
        vo.setDataRows(JSONObject.parseObject(bduKafkaMsg.getNr(), List.class));
        return vo;
    }

}
